package com.springboot.zookeeper;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import javax.annotation.PostConstruct;

import org.I0Itec.zkclient.IZkDataListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheEvent;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Configuration;

import com.springboot.common.util.StringUtil;
import com.springboot.config.ConsumerConfig;

import lombok.extern.log4j.Log4j2;

/**
 * zk配置参数服务
 *
 * @author supu
 * @create 2018-12-27 19:24
 **/
@Log4j2
@Configuration
public class ZookeeperConfigService {
    private static final Map<String,String> DATA_MAP = new HashMap<>(16);

    @Autowired
    private ConsumerConfig consumerConfig;
    @Autowired
    private ZkClient zkClient;
    @Autowired
    private CuratorFramework curatorFramework;

    /**
     * 节点数据变动监听器
     * 此种方式不支持节点增加的事件监听
     */
    private final IZkDataListener listener = new IZkDataListener() {
        @Override
        public void handleDataChange(String dataPath, Object data) throws Exception {
            String key = dataPath.substring(dataPath.lastIndexOf("/") + 1);
            log.warn("dataPath:{}, key: {} 对应的 value:{}", dataPath, key, data);
            DATA_MAP.put(key, data.toString());
            log.warn(DATA_MAP);
        }

        @Override
        public void handleDataDeleted(String dataPath) throws Exception {
            String key = dataPath.substring(dataPath.lastIndexOf("/") + 1);
            log.warn("dataPath:{} 对应的 key: {} ", dataPath, key);
            DATA_MAP.remove(key);
            log.warn(DATA_MAP);
        }
    };

    @PostConstruct
    public void initZookeeperConfig() {
        log.info("init start");

        String rootPath = consumerConfig.getString(ConsumerConfig.ZOOKEEPER_GROUP);
        log.info("zkclient query start");
//        initZkClientQueryData(rootPath);
        log.info("zkclient query start");

        log.info("curator query start");
        initCuratorQueryData(rootPath);
        log.info("curator query end");

        // 初始化zookeeper监听器
//        initDataWatcher(rootPath);
//        curatorStartTreeCacheWatch(rootPath);
//        curatorStartPathChildrenWatch(rootPath);
//        curatorStartNodeChildrenWatch(rootPath);
    }

    private void initZkClientQueryData(String path) {
        List<String> childrenList = zkClient.getChildren(path);
        childrenList.forEach(children -> {
            String childPath = path + "/" + children;
            List<String> children1List = zkClient.getChildren(childPath);
            if (children1List == null || children1List.size() == 0) {
                log.warn("key: {} 对应的 value:{}", children, zkClient.readData(childPath, true));
                DATA_MAP.put(children, zkClient.readData(childPath, true));
            } else {
                initZkClientQueryData(childPath);
            }
        });
    }


    public void initDataWatcher(String path) {
        List<String> childrenList = zkClient.getChildren(path);

        childrenList.forEach(children -> {
            String childPath = path + "/" + children;
            List<String> children1List = zkClient.getChildren(childPath);
            if (children1List == null || children1List.size() == 0) {
                zkClient.subscribeDataChanges(childPath, listener);
            } else {
                initDataWatcher(childPath);
            }
        });
    }

    public void initWatcher() {
        String watchNodePath = consumerConfig.getString(ConsumerConfig.ZOOKEEPER_GROUP) + "/watch_test";

        zkClient.subscribeChildChanges(watchNodePath, (s, list) -> {
            list.forEach(node -> {
                String childPath = watchNodePath + "/" + node;
                String value = zkClient.readData(childPath, true);
                log.warn("子节点: {} 对应的 data:{}", node, value);

                try {
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    log.warn("sleep 5s");
                }

                if (zkClient.exists(childPath)) {
                    zkClient.delete(childPath);
                }
            });
        });
    }

    /**
     * 使用curator客户端，监听某路径下的所有节点数据变化，包含增删改
     *
     * @param path
     */
    private void curatorStartTreeCacheWatch(String path) {
        TreeCache cache = new TreeCache(curatorFramework, path);
        try {
            cache.getListenable().addListener((curatorFramework, event) -> {
                ChildData childData = event.getData();
                if(childData != null){
                    String dataPath = childData.getPath();
                    String key = dataPath.substring(dataPath.lastIndexOf("/") + 1);
                    String data = new String(childData.getData());
                    log.warn("事件类型：{}, path:{}, key:{}, data：{}", event.getType(), dataPath, key, data);
                    if (TreeCacheEvent.Type.NODE_ADDED == event.getType() || TreeCacheEvent.Type.NODE_UPDATED == event.getType()) {
                        if (StringUtil.isNotEmpty(data)) {
                            DATA_MAP.put(key, data);
                            log.warn(DATA_MAP);
                        }
                    } else if (TreeCacheEvent.Type.NODE_REMOVED == event.getType()) {
                        if (StringUtil.isNotEmpty(data)) {
                            DATA_MAP.remove(key);
                            log.warn(DATA_MAP);
                        }
                    }
                }
            });
            cache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 使用curator客户端，监听某路径下的所有节点数据变化，包含增删改
     * 此方法已经过期了
     * 当删除到只剩下节点时，此方法有bug了，删除之后同时触发了删除和增加事件
     *
     * @param path
     */
    private void curatorStartPathChildrenWatch(String path) {
        PathChildrenCache watch = new PathChildrenCache(curatorFramework, path, true);

        watch.getListenable().addListener((curatorFramework, event) -> {
            ChildData childData = event.getData();
            if(childData != null){
                String dataPath = childData.getPath();
                String key = dataPath.substring(dataPath.lastIndexOf("/") + 1);
                String data = new String(childData.getData());
                log.warn("事件: {}, dataPath:{}, key:{}, data:{}", event.getType(), dataPath, key, data);
                if (PathChildrenCacheEvent.Type.CHILD_ADDED == event.getType()) {
                    if (StringUtil.isNotEmpty(data)) {
                        DATA_MAP.put(key, data);
                    } else {
                        curatorStartPathChildrenWatch(dataPath);
                    }
                } else if (PathChildrenCacheEvent.Type.CHILD_REMOVED == event.getType()) {
                    DATA_MAP.remove(key);
                } else if (PathChildrenCacheEvent.Type.CHILD_UPDATED == event.getType()) {
                    DATA_MAP.put(key, data);
                }
                log.warn(DATA_MAP);
            }
        });

        try {
            watch.start(PathChildrenCache.StartMode.BUILD_INITIAL_CACHE);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 使用curator客户端，监听某个具体节点数据变化，包含增删改
     * 此方法已经过期了
     *
     * @param path
     */
    private void curatorStartNodeChildrenWatch(String path) {
        path += "/oo";
        NodeCache nodeCache = new NodeCache(curatorFramework, path);

        nodeCache.getListenable().addListener(() -> {
            ChildData childData = nodeCache.getCurrentData();
            String dataPath = childData == null ? nodeCache.getPath() : childData.getPath();
            String key = dataPath.substring(dataPath.lastIndexOf("/") + 1);
            if (childData == null) {
                // 节点被删除了
                log.warn("key:{} 被删除了", key);
                DATA_MAP.remove(key);
            } else {
                String data = new String(childData.getData());
                log.warn("dataPath:{}, key:{}, data:{}", dataPath, key, data);
                DATA_MAP.put(key, data);
            }
            log.warn(DATA_MAP);
        });

        try {
            nodeCache.start();
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    private void initCuratorQueryData(String path) {
        try {
            List<String> childList = curatorFramework.getChildren().forPath(path);
            childList.forEach(node -> {
                try {
                    String childPath = path + "/" + node;
                    List<String> childTreeList = curatorFramework.getChildren().forPath(childPath);
                    if (childTreeList == null || childTreeList.size() == 0) {
                        byte[] bytes = curatorFramework.getData().forPath(childPath);
                        String data = new String(bytes);
                        log.warn("key: {} 对应的 value:{}", node, data);
                        DATA_MAP.put(node, data);
                    } else {
                        initCuratorQueryData(childPath);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public String getValue(String key) {
        return DATA_MAP.get(key);
    }
}
